package com.zhf.service;

import com.zhf.config.KafkaConfig;
import com.zhf.serializer.GsonSerializer;
import com.zhf.serializer.ISerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Properties;

@Component
public class KafkaPublishService {

    private KafkaProducer<String, byte[]> producer;

    @Autowired
    KafkaConfig config;

    ISerializer serializer = GsonSerializer.getInstance();

    @PostConstruct
    public void initializePublisher() {
        Properties properties = config.initKafkaProperties();
        producer = new KafkaProducer<>(properties);
    }

    @PreDestroy
    public void clean() {
        producer.flush();
        producer.close();
    }

    private void send(String topic, String key, byte[] value) {
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(config.getKafkaTopic(), null, value);
        producer.send(record);
    }

    private void send(String topic, byte[] value) {
        send(topic, null, value);
    }

    public void publish(Object obj) {
        byte[] bytes = serializer.toByteArray(obj);
        send(config.getKafkaTopic(), bytes);
        producer.flush();
    }

}
